package top.byteio.common.mq.factory;

import com.alibaba.fastjson2.JSON;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import top.byteio.common.cache.lock.DistributedLock;
import top.byteio.common.core.constants.MqConstants;
import top.byteio.common.core.utils.StringUtils;
import top.byteio.common.log.annotation.WebLog;
import top.byteio.common.log.enums.LogEnum;
import top.byteio.common.mq.exception.MqException;
import top.byteio.common.mq.handler.MqProducerSender;
import top.byteio.common.mq.pojo.dto.MqDTO;
import top.byteio.common.mq.pojo.dto.MqResultDTO;

import java.util.concurrent.TimeUnit;

/**
* rabbitmq消息工厂
* @author lerryxia
*/
public abstract class AbstractRabbitMqFactory<T> extends AbstractMqFactory<T> {


    public AbstractRabbitMqFactory(DistributedLock distributedLock,MqProducerSender mqProducerSender) {
        super(distributedLock,mqProducerSender);
    }

    /**
     * 消息生产
     *
     * @param mqDto 消息体
     * @return 结果
     * @throws MqException 异常
     */
    @Override
    @WebLog(msg = "RabbitMq sen message",type = LogEnum.MQ)
    public MqResultDTO sendMq(MqDTO<T> mqDto) throws MqException {
        return mqProducerSender.sendMq(mqDto);
    }

    /**
     * 消息监听,消费消息
     *
     * @param value 消息体
     * @throws MqException 异常
     */
    @Override
    @RabbitHandler
    @WebLog(msg = "RabbitMq listener message",type = LogEnum.MQ)
    protected void listener(String value) throws MqException {
        if (StringUtils.isEmpty(value)) {
            throw new MqException("RabbitMq listener value isEmpty ");
        }
        MqDTO<T> mqDto = JSON.parseObject(value,  MqDTO.class);
        //分布式锁key
        String key = StringUtils.format(MqConstants.LOCK_KEY, mqDto.getTopic() + "_" + mqDto.getPriKey());
        try {
            //获取分布式锁
            if (distributedLock.tryLock(key, MqConstants.LOCK_TIMEOUT, TimeUnit.SECONDS)) {
                //业务消费处理
                consumeMq(mqDto);
            }
        } catch (Exception e) {
            throw new MqException(e.getMessage());
        } finally {
            distributedLock.unLock(key);
        }
    }
}
